package com.zy.flink.dataset.datastream;




import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHouseDataSource;
import ru.yandex.clickhouse.ClickHouseStatement;
import ru.yandex.clickhouse.settings.ClickHouseProperties;
import ru.yandex.clickhouse.settings.ClickHouseQueryParam;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * @author Dell
 * @Title:
 * @Description: 请写注释类
 * @date 2024/1/18
 */
class ClickHouseSource extends RichParallelSourceFunction<Map<String, Object>> {

    private boolean flag = true;
    private ClickHouseConnection conn = null;
    private ClickHouseStatement stmt = null;
    private ResultSet rs = null;
    private Map<ClickHouseQueryParam, String> additionalDBParams = new HashMap<>();

    // open只执行一次,适合开启资源
    @Override
    public void open(Configuration parameters) throws Exception {
        String url = "jdbc:clickhouse://192.168.215.17:8123";

        ClickHouseProperties properties = new ClickHouseProperties();
        properties.setSessionId(UUID.randomUUID().toString());
//        properties.setUser("default");
//        properties.setPassword("");
        properties.setDatabase("beluga");

        ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);
        // ClickHouseProperties
        additionalDBParams.put(ClickHouseQueryParam.SESSION_ID, UUID.randomUUID().toString());
        conn = dataSource.getConnection();
        stmt = conn.createStatement();
    }

    @Override
    public void run(SourceContext<Map<String, Object>> ctx) throws Exception {

        String sql = " select * from src_soc_alarm_log_all where received_time >= 1702436881610 limit %s,%s";
        String start = "2023-9-01 00:00:00";
        String end = "2023-9-02 23:59:59";
        Date date = new Date();

        int limitStart = 0;
        int limit = 10;
//        while (flag) {
            String newSql = String.format(sql,  limitStart, limit);

            rs = stmt.executeQuery(newSql, additionalDBParams);
//            rs = stmt.executeQuery(newSql);
//            Map<String, Object> hm = Maps.newHashMap();
//            ResultSetMetaData rsmd = rs.getMetaData();
//            int count = rsmd.getColumnCount();// 获取列的数量
//            for (int i = 1; i <= count; i++) {
//                String key = rsmd.getColumnLabel(i);
//
//                Object value = rs.getObject(i);
//                hm.put(key, value);
//            }
//            ctx.collect(hm);
//            int c = 0;
            while (rs.next()) {
//                c++;
                Map<String, Object> hm = Maps.newHashMap();
                ResultSetMetaData rsmd = rs.getMetaData();
                int count = rsmd.getColumnCount();// 获取列的数量
                for (int i = 1; i <= count; i++) {
                    String key = rsmd.getColumnLabel(i);
                    Object value = rs.getObject(i);
                    hm.put(key, value);
                }
                ctx.collect(hm);
            }

//            limitStart += c;

//            if (c < limit) {
//                // 查完了，换时间
//                limitStart = 0;
//                LocalDate localDate = LocalDate.parse(start, DateTimeFormatter.ofPattern("yyyy-MM-dd")).plusDays(1);
//                start = localDate.toString();
//                end = localDate.toString();
//
//                if (localDate.isAfter(LocalDate.now())) {
//                    flag = false;
//                    return;
//                }
//            }

            // 按需处理
//            Thread.sleep(1000 * 10);
//        }
    }

    // 接收到cancel命令时取消数据生成
    @Override
    public void cancel() {
        flag = false;
    }

    @Override
    public void close() throws Exception {
        if (conn != null)
            conn.close();
        if (stmt != null)
            stmt.close();
        if (rs != null)
            rs.close();
    }

}
        